package com.cy.improve.demos.web.mq;


import com.alibaba.fastjson2.JSON;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.TopicPartitionInfo;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

@Component
public class MessageProducer {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    public void sendMessage(String topic, String message) {
        kafkaTemplate.send(topic, message);
    }

    // 创建topic的接口，接收topic名称、分区数量和副本数量作为参数
    public ResponseEntity<String> createTopic(String topicName, int partitions, int replicas) {
        // 配置KafkaAdminClient连接属性
        Properties properties = new Properties();
        properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);

        try (AdminClient adminClient = AdminClient.create(properties)) {
            // 创建NewTopic对象，指定topic名称、分区数量和副本数量
            NewTopic newTopic = new NewTopic(topicName, partitions, (short) replicas);
            // 使用AdminClient创建topic
            adminClient.createTopics(Collections.singleton(newTopic)).all().get();
            return new ResponseEntity<>("Topic " + topicName + " created successfully.", HttpStatus.OK);
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
            return new ResponseEntity<>("Failed to create topic: " + e.getMessage(), HttpStatus.INTERNAL_SERVER_ERROR);
        }
    }

    public List<Map<String, Object>> getTopicDetails() throws ExecutionException, InterruptedException {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", bootstrapServers);

        try (AdminClient adminClient = AdminClient.create(properties)) {
            ListTopicsResult listTopicsResult = adminClient.listTopics();
            Collection<TopicListing> topicListings = listTopicsResult.listings().get();

            List<Map<String, Object>> topicDetailsList = new ArrayList<>();

            for (TopicListing topicListing : topicListings) {
                String topicName = topicListing.name();
                DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Collections.singleton(topicName));
                Map<String, TopicDescription> topicDescriptions = describeTopicsResult.all().get();
                TopicDescription topicDescription = topicDescriptions.get(topicName);

                Map<String, Object> topicDetails = new HashMap<>();
                topicDetails.put("topicName", topicName);
                topicDetails.put("partitions", topicDescription.partitions().size());
                List<Node> replicas = new ArrayList<>();
                for (TopicPartitionInfo partitionInfo : topicDescription.partitions()) {
                    replicas.addAll(partitionInfo.replicas());
                }
                topicDetails.put("replicas", new HashSet<>(replicas));
                topicDetailsList.add(topicDetails);
            }

            return topicDetailsList;
        }
    }


    public void deleteTopic(String topicName) throws ExecutionException, InterruptedException {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", bootstrapServers);

        try (AdminClient adminClient = AdminClient.create(properties)) {
            DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(Collections.singleton(topicName));
            deleteTopicsResult.all().get();
        }
    }


    // @Autowired
    // private ProducerFactory<String, String> producerFactory;


    @Autowired
    private Producer producer;


    public void sendMessagesBatch(List<String> messages, String topic) {


        for (String message : messages) {

            ProducerRecord producerRecord = new ProducerRecord(topic, message);
            producer.send(producerRecord);

        }
    }


}

